package ex.tableapi.join;

import ex.tableapi.ApiFrame;
import org.apache.flink.table.api.Table;

import static org.apache.flink.table.api.Expressions.$;

public class ExInnerJoin extends ApiFrame {
    private String outName = "printOutTable";

    public static void main(String[] args) {
        ExInnerJoin demo = new ExInnerJoin();
        demo.getEnv();
        Table orders = demo.registerMysql("myorder", "orders");

        Table result1 = demo.exInnerJoin(orders);
        demo.registerConsole(demo.createPrintOutDDL(result1.getResolvedSchema().toString()), demo.outName);
        result1.executeInsert(demo.outName);
    }
    private Table exInnerJoin(Table orders) {
        Table left = orders.select($("a"), $("b"), $("c"));
        Table right = orders.select($("d"), $("e"), $("f"));
        Table result = left.join(right)
                .where($("a").isEqual($("d")))
                .select($("a"), $("b"), $("e"));
        return result;
    }


}
